package com.bbx.iotdb.iotdbdemo.mqtt;


import com.alibaba.fastjson.JSON;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;
import lombok.extern.slf4j.Slf4j;

import java.util.List;


@Slf4j
public class MQTTSubscribeClient {

    public static void main(String[] args) {
        Mqtt3AsyncClient client = Mqtt3Client.builder()
                .identifier("bbbbbb")
                .serverPort(2883)
                .serverHost("192.168.10.85")
                .buildAsync();

        client.connectWith()
                .simpleAuth()
                .username("root")
                .password("root".getBytes())
                .applySimpleAuth()
                //是否指定会话保持，指定hui'hua
//                .cleanSession(false)
                .send()
                .whenComplete((connAck, throwable) -> {
                    if (throwable != null) {
                        log.info("[{}]", "连接失败");

                    } else {
                        log.info("[{}]", "连接成功");
                        client.subscribeWith()
                                .topicFilter("ddl/topic")
                                //服务质量，订阅消息时可以不指定，默认为 AT_LEAST_ONCE
//                                .qos(MqttQos.AT_LEAST_ONCE)
                                .callback(publish -> {
                                    // Process the received message
                                    String s = new String(publish.getPayloadAsBytes());
                                    PayloadEntity payloadEntity = JSON.parseObject(s, PayloadEntity.class);
                                    log.info("payload[{}]", payloadEntity);
                                    String device = payloadEntity.getDevice();
                                    List<Long> timestamps = payloadEntity.getTimestamps();
                                    List<List<Object>> values = payloadEntity.getValues();
                                    log.info("device[{}],timestamps[{}],values[{}]", device, timestamps.get(0), values.get(0));
                                })
                                .send();
                    }


                });
        //取消订阅
        //client.unsubscribeWith().topicFilter("ddl/topic").send();

        client.disconnect();


    }
}
